use std::sync::Arc;
use chrono::{DateTime, Utc, TimeZone};
use sqlx::{Pool, Sqlite, QueryBuilder};
use anyhow::{Result, anyhow};
use serde::{Deserialize, Serialize};
use rust_decimal::prelude::*;

use crate::models::stock_candle::StockCandle;

/// 安全地将 Decimal 转换为 f64
fn decimal_to_f64(decimal: Decimal) -> Result<f64> {
    decimal.to_f64().ok_or_else(|| anyhow!("Failed to convert Decimal to f64"))
}

/// 安全地将 Option<Decimal> 转换为 Option<f64>
fn optional_decimal_to_f64(decimal: Option<Decimal>) -> Result<Option<f64>> {
    decimal.map(decimal_to_f64).transpose()
}

/// K线查询参数
#[derive(Debug, Clone, Deserialize)]
pub struct CandleQueryParams {
    pub symbol: String,
    pub period: String,
    pub start_time: Option<DateTime<Utc>>,
    pub end_time: Option<DateTime<Utc>>,
    pub limit: Option<i64>,
}

/// 市场概览数据
#[derive(Debug, Serialize)]
pub struct MarketSummary {
    pub total_stocks: i64,
    pub up_count: i64,
    pub down_count: i64,
    pub unchanged_count: i64,
    pub turnover_amount: f64,
    pub turnover_volume: f64,
    pub timestamp: DateTime<Utc>,
}

/// 股票K线服务
pub struct StockCandleService {
    pool: Arc<Pool<Sqlite>>,
}

impl StockCandleService {
    /// 创建新的股票K线服务实例
    pub fn new(pool: Arc<Pool<Sqlite>>) -> Self {
        Self { pool }
    }
    
    /// 查询K线数据
    pub async fn query_candles(&self, params: CandleQueryParams) -> Result<Vec<StockCandle>> {
        let mut query_builder = QueryBuilder::new("SELECT * FROM stock_candles WHERE symbol = ");
        query_builder.push_bind(params.symbol.clone());
        query_builder.push(" AND period = ");
        query_builder.push_bind(params.period.clone());
        
        if let Some(start_time) = params.start_time {
            query_builder.push(" AND timestamp >= ");
            query_builder.push_bind(start_time);
        }
        
        if let Some(end_time) = params.end_time {
            query_builder.push(" AND timestamp <= ");
            query_builder.push_bind(end_time);
        }
        
        query_builder.push(" ORDER BY timestamp DESC");
        
        if let Some(limit) = params.limit {
            query_builder.push(" LIMIT ");
            query_builder.push_bind(limit);
        }
        
        let query = query_builder.build_query_as::<StockCandle>();
        let candles = query.fetch_all(&*self.pool).await?;
        Ok(candles)
    }
    /// 获取市场概览数据
    pub async fn get_market_summary(&self) -> Result<MarketSummary> {
        // 使用 query! 宏并指定正确的类型
        let row = sqlx::query!(
            r#"
            WITH latest_timestamps AS (
                SELECT symbol, MAX(timestamp) as max_timestamp
                FROM stock_candles
                WHERE period = '1d'
                GROUP BY symbol
            ),
            latest_prices AS (
                SELECT 
                    sc.symbol,
                    sc.close,
                    sc.volume,
                    sc.amount,
                    sc.timestamp,
                    (
                        SELECT sc2.close
                        FROM stock_candles sc2
                        WHERE sc2.symbol = sc.symbol
                        AND sc2.period = '1d'
                        AND sc2.timestamp < sc.timestamp
                        ORDER BY sc2.timestamp DESC
                        LIMIT 1
                    ) as prev_close
                FROM stock_candles sc
                JOIN latest_timestamps lt ON sc.symbol = lt.symbol AND sc.timestamp = lt.max_timestamp
                WHERE sc.period = '1d'
            )
            SELECT 
                COUNT(*) as total_stocks,
                COALESCE(SUM(CASE WHEN close > prev_close THEN 1 ELSE 0 END), 0) as up_count,
                COALESCE(SUM(CASE WHEN close < prev_close THEN 1 ELSE 0 END), 0) as down_count,
                COALESCE(SUM(CASE WHEN close = prev_close OR prev_close IS NULL THEN 1 ELSE 0 END), 0) as unchanged_count,
                COALESCE(SUM(CAST(amount AS REAL)), 0) as turnover_amount,
                COALESCE(SUM(CAST(volume AS REAL)), 0) as turnover_volume,
                COALESCE(MAX(timestamp), CURRENT_TIMESTAMP) as timestamp
            FROM latest_prices
            "#
        )
        .fetch_one(&*self.pool)
        .await?;
        
        // 解析时间戳字符串为 DateTime<Utc>
        let timestamp = {
            // 如果时间戳为 None，则使用当前时间
            if let Some(ts) = row.timestamp {
                // 尝试解析 SQLite 时间戳格式 (例如: "2023-06-15 10:30:00")
                let parsed_result = DateTime::parse_from_str(&format!("{} +0000", ts), "%Y-%m-%d %H:%M:%S %z")
                    .or_else(|_| DateTime::parse_from_str(&format!("{}Z", ts), "%Y-%m-%dT%H:%M:%S%.fZ"))
                    .or_else(|_| DateTime::parse_from_str(&ts, "%Y-%m-%d %H:%M:%S%.f"));
                
                match parsed_result {
                    Ok(dt) => dt.with_timezone(&Utc),
                    Err(_) => Utc::now()
                }
            } else {
                Utc::now()
            }
        };
        
        Ok(MarketSummary {
            total_stocks: row.total_stocks as i64,
            up_count: row.up_count as i64,
            down_count: row.down_count as i64,
            unchanged_count: row.unchanged_count as i64,
            turnover_amount: row.turnover_amount as f64,
            turnover_volume: row.turnover_volume as f64,
            timestamp,
        })
    }
    
    /// 批量保存K线数据
    pub async fn save_candles(&self, candles: Vec<StockCandle>) -> Result<()> {
        let mut tx = self.pool.begin().await?;
        
        for candle in candles {
            // 先转换所有的 Decimal 值，以便在出错时提供更具体的错误信息
            let open = decimal_to_f64(candle.open)
                .map_err(|e| anyhow!("Failed to convert open price for {}: {}", candle.symbol, e))?;
            let high = decimal_to_f64(candle.high)
                .map_err(|e| anyhow!("Failed to convert high price for {}: {}", candle.symbol, e))?;
            let low = decimal_to_f64(candle.low)
                .map_err(|e| anyhow!("Failed to convert low price for {}: {}", candle.symbol, e))?;
            let close = decimal_to_f64(candle.close)
                .map_err(|e| anyhow!("Failed to convert close price for {}: {}", candle.symbol, e))?;
            let volume = decimal_to_f64(candle.volume)
                .map_err(|e| anyhow!("Failed to convert volume for {}: {}", candle.symbol, e))?;
            let amount = optional_decimal_to_f64(candle.amount)
                .map_err(|e| anyhow!("Failed to convert amount for {}: {}", candle.symbol, e))?;
            let change_percent = optional_decimal_to_f64(candle.change_percent)
                .map_err(|e| anyhow!("Failed to convert change_percent for {}: {}", candle.symbol, e))?;
            let turnover_rate = optional_decimal_to_f64(candle.turnover_rate)
                .map_err(|e| anyhow!("Failed to convert turnover_rate for {}: {}", candle.symbol, e))?;
            
            sqlx::query!(
                r#"
                INSERT INTO stock_candles (
                    symbol, period, timestamp, open, high, low, close, 
                    volume, amount, change_percent, turnover_rate, 
                    is_trading_day, source, created_at, updated_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT (symbol, period, timestamp) 
                DO UPDATE SET
                    open = excluded.open,
                    high = excluded.high,
                    low = excluded.low,
                    close = excluded.close,
                    volume = excluded.volume,
                    amount = excluded.amount,
                    change_percent = excluded.change_percent,
                    turnover_rate = excluded.turnover_rate,
                    is_trading_day = excluded.is_trading_day,
                    source = excluded.source,
                    updated_at = datetime('now')
                "#,
                candle.symbol,
                candle.period,
                candle.timestamp,
                open,
                high,
                low,
                close,
                volume,
                amount,
                change_percent,
                turnover_rate,
                candle.is_trading_day,
                candle.source,
                candle.created_at,
                candle.updated_at,
            )
            .execute(&mut *tx)
            .await?;
        }
        
        tx.commit().await?;
        Ok(())
    }
    
    /// 删除指定股票的K线数据
    pub async fn delete_candles(&self, symbol: &str, period: Option<&str>) -> Result<u64> {
        let mut query_builder = QueryBuilder::new("DELETE FROM stock_candles WHERE symbol = ");
        query_builder.push_bind(symbol);
        
        if let Some(period) = period {
            query_builder.push(" AND period = ");
            query_builder.push_bind(period);
        }
        
        let query = query_builder.build();
        let result = query.execute(&*self.pool).await?;
        
        Ok(result.rows_affected())
    }
}